package com.colorlight.cloud.mqtt.base.core.connect;

import com.colorlight.cloud.common.exception.BaseException;
import com.colorlight.cloud.mqtt.base.api.setting.ConnectSetting;
import com.colorlight.cloud.mqtt.base.core.properties.MqttProperties;
import com.colorlight.cloud.mqtt.base.core.tls.*;
import com.colorlight.cloud.mqtt.base.entity.*;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.*;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.net.ssl.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Set;


/**
 * @author Demon
 * 连接broker 并配置相关主题
 * 这里的实现有问题 spring继承的mqttv5 或者mqttv3 都不能满足 一个clientID既监听又下发
 */
@Configuration
@IntegrationComponentScan
@Slf4j
public class MqttConnect {
	
	private static final String INPUT_SUFFIX = "INPUT";
	private static final String OUTPUT_SUFFIX = "OUTPUT";
	
	private final ConnectSetting connectSetting;
	
	private final MessageHandler messageHandler;
	
	private final String url;
	private final String username;
	private final String password;
	
	private final String inputClientId;
	private final String outputClientId;
	private final int[] qos;
	private final String[] topics;
	
	private final MqttProperties properties;
	
	private final MySslContext mySslContext;
	
	public MqttConnect(ConnectSetting connectSetting, @Qualifier("MessageHandler") MessageHandler messageHandler, MqttProperties properties) {
		this.connectSetting = connectSetting;
		this.messageHandler = messageHandler;
		this.properties = properties;
		ConnectDetail connectDetail = connectSetting.getConnectDetail();
		url = connectDetail.getBrokerUrl();
		username = connectDetail.getUsername();
		password = connectDetail.getPassword();
		String clientId = connectDetail.getClientId();
		inputClientId = clientId + INPUT_SUFFIX;
		outputClientId = clientId + OUTPUT_SUFFIX;
		Set<TopicDetail> topicDetails = connectDetail.getTopicDetails();
		
		topics = topicDetails.stream().map(TopicDetail::getTopic).toArray(String[]::new);
		qos = topicDetails.stream().map(TopicDetail::getQosEnum).map(QosEnum::getValue).mapToInt(Integer::intValue).toArray();
		mySslContext = new MySslContext();
	}
	
	private MqttConnectOptions getMqttConnectOptions() {
		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
		//是否清除会话
		mqttConnectOptions.setCleanSession(false);
		mqttConnectOptions.setAutomaticReconnect(true);
		mqttConnectOptions.setUserName(username);
		mqttConnectOptions.setPassword(
				password.toCharArray());
		mqttConnectOptions.setServerURIs(
				new String[]{url});
		//心跳
		mqttConnectOptions.setKeepAliveInterval(30);
		//超时时间
		mqttConnectOptions.setConnectionTimeout(30);
		if (Boolean.TRUE.equals(properties.getTls())) {
			//SSL认证
			SSLContext sslContext = null;
			try {
				sslContext = mySslContext.createSSLContext(properties.getClientCertPath(), properties.getClientKeyPath(), properties.getCaCertPath());
			} catch (Exception e) {
				//受检异常 捕获 抛出异常   不可以做处理 防止重连
				throw new BaseException("证书加载异常");
			}
			SSLSocketFactory sslSocketFactory = null;
			String[] strings = splitUri(url);
			Objects.requireNonNull(strings, "服务地址异常");
			String hostname = strings[0];
			int port = 8883;
			if (strings[1] != null) {
				port = Integer.parseInt(strings[1]);
			}
			try {
				assert sslContext != null;
				sslSocketFactory = sslContext.getSocketFactory();
				SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket();
				sslSocket.connect(new InetSocketAddress(hostname, port));
			} catch (Exception e) {
				log.error("",e);
			}
			mqttConnectOptions.setSocketFactory(sslSocketFactory);
		}
		return mqttConnectOptions;
	}
	
	/**
	 * 客户端状态存储： 将客户端的订阅、未确认的消息等状态信息保存到本地文件系统。
	 * 这使得客户端能够在重新启动后恢复其订阅和其他状态，而不会丢失先前的信息。
	 * <p>
	 * 消息持久性： 对于 QoS (Quality of Service) 为 1 或 2 的消息，MqttDefaultFilePersistence 确保它们在发送后被持久化。
	 * 这样，如果客户端在消息发送后崩溃或重新启动，可以确保这些消息能够被送达
	 */
	private MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		factory.setConnectionOptions(getMqttConnectOptions());
		//停止发布后的订阅策略
		factory.setConsumerStopAction(ConsumerStopAction.UNSUBSCRIBE_NEVER);
		
		if (Objects.nonNull(connectSetting.getMqttClientPersistence())) {
			factory.setPersistence(connectSetting.getMqttClientPersistence());
		}
		
		return factory;
	}
	
	
	@Bean
	public MqttPahoMessageDrivenChannelAdapter adapter() {
		for (String topic : topics) {
			log.info("当前监听主题:{}", topic);
		}
		MqttPahoMessageDrivenChannelAdapter adapter;
		//TOPICS 底层 LinkedHashSet
		adapter = new MqttPahoMessageDrivenChannelAdapter(
				inputClientId,
				mqttClientFactory(),
				topics
		);
		
		
		//StandardCharsets.ISO_8859_1.toString()
		adapter.setConverter(
				new DefaultPahoMessageConverter(
						2,
						true,
						StandardCharsets.ISO_8859_1.toString()));
		
		adapter.setOutputChannel(mqttInputChannel());
		adapter.setErrorChannel(null);
		adapter.setQos(qos);
		return adapter;
	}
	
	
	/**
	 * ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
	 * 对添加的主题进行监听
	 **/
	@Bean
	@ServiceActivator(inputChannel = "InputMessageChannel")
	public MessageHandler handler() {
		return messageHandler;
	}
	
	
	@Bean
	@ServiceActivator(inputChannel = "mqttOutboundChannel")
	public MessageHandler mqttOutbound() {
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(outputClientId, mqttClientFactory());
		//发送消息时将不会阻塞。
		messageHandler.setAsync(true);
		messageHandler.setDefaultQos(1);
		return messageHandler;
	}
	
	@Bean
	public MessageChannel mqttOutboundChannel() {
		return new DirectChannel();
	}
	
	@Bean(name = "InputMessageChannel")
	public MessageChannel mqttInputChannel() {
		return new DirectChannel();
	}
	
	
	private static String[] splitUri(String uriString) {
		try {
			URI uri = new URI(uriString);
			String host = uri.getHost();
			int port = uri.getPort();
			if (host != null && port != -1) {
				return new String[]{host, String.valueOf(port)};
			}
		} catch (URISyntaxException e) {
			throw new BaseException("MQTT地址解析异常");
		}
		throw new BaseException("MQTT地址解析异常");
	}
}
